package com.networkflow_analysis;

import com.networkflow_analysis.bean.PageViewCount;
import com.networkflow_analysis.bean.UserBehavior;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.TimeCharacteristic;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.timestamps.AscendingTimestampExtractor;
import org.apache.flink.streaming.api.functions.windowing.ProcessAllWindowFunction;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.triggers.Trigger;
import org.apache.flink.streaming.api.windowing.triggers.TriggerResult;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;
import redis.clients.jedis.Jedis;

/**
 * @Description: TODO QQ1667847363
 * @author: xiao kun tai
 * @date:2021/11/10 23:54
 *
 * 使用布隆过滤器的 UV 统计
 */
public class UvWithBloomFilter {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime);
        env.setParallelism(1);

        String filePath = "NetworkFlowAnalysis/src/main/resources/UserBehavior.csv";
        DataStream<String> fileStream = env.readTextFile(filePath);

        //转换为POJO，分配时间戳和watermark
        DataStream<UserBehavior> dataStream = fileStream
                .map(line -> {
                    String[] fields = line.split(",");
                    return new UserBehavior(new Long(fields[0]), new Long(fields[1]), new Integer(fields[2]), fields[3], new Long(fields[4]));
                })
                /**
                 * 有序窗口
                 */
                .assignTimestampsAndWatermarks(new AscendingTimestampExtractor<UserBehavior>() {
                    @Override
                    public long extractAscendingTimestamp(UserBehavior userBehavior) {
                        return userBehavior.getTimestamp() * 1000L;
                    }
                });
        /**
         * 开窗统计uv值
         */
        SingleOutputStreamOperator<PageViewCount> uvStream = dataStream
                .filter(data -> "pv".equals(data.getBehavior()))
                .timeWindowAll(Time.hours(1))
                .trigger(new MyTrigger())
                .process(new UvCountResultWithBloomFilter());

        uvStream.print();

        env.execute("uv count wth bloom filter job");
    }

    /**
     * 自定义触发器
     * OnElement ：每一个数据进入窗口都会触发。
     * OnEventTime ：根据接入窗口的EventTime进行触发操作
     * OnProcessTime ： 根据接入窗口的ProcessTime进行触发操作
     * Clear ： 执行窗口及状态数据的清除方法。
     */
    public static class MyTrigger extends Trigger<UserBehavior, TimeWindow> {
        @Override
        public TriggerResult onElement(UserBehavior userBehavior, long l, TimeWindow timeWindow, TriggerContext triggerContext) throws Exception {
            /**
             * 每一条数据来到，直接触发窗口极端，并且直接清空窗口
             * CONTINUE ： 不进行操作，等待。
             * FIRE ： 触发计算且数据保留。
             * PRUGE ： 窗口内部数据清除且不触发计算。
             * FIRE_AND_PURGE : 触发计算并清除对应的数据。
             */
            return TriggerResult.FIRE_AND_PURGE;
        }

        @Override
        public TriggerResult onProcessingTime(long l, TimeWindow timeWindow, TriggerContext triggerContext) throws Exception {
            return TriggerResult.CONTINUE;
        }

        @Override
        public TriggerResult onEventTime(long l, TimeWindow timeWindow, TriggerContext triggerContext) throws Exception {
            return TriggerResult.CONTINUE;
        }

        @Override
        public void clear(TimeWindow timeWindow, TriggerContext triggerContext) throws Exception {

        }
    }

    //自定义一个布尔过滤器
    public static class MyBloomFilter {
        //定义位图的大小，一般定义为2的整次幂
        private Integer cap;

        public MyBloomFilter(Integer cap) {
            this.cap = cap;
        }

        //实现一个Hash函数
        public Long hashCode(String value, Integer seed) {
            Long result = 0L;
            for (int i = 0; i < value.length(); i++) {
                result = result * seed + value.charAt(i);
            }
            return result & (cap - 1);
        }
    }

    /**
     * 实现自定义处理函数
     */
    public static class UvCountResultWithBloomFilter extends ProcessAllWindowFunction<UserBehavior, PageViewCount, TimeWindow> {

        //定义jedis连接和布隆过滤器
        Jedis jedis;
        MyBloomFilter myBloomFilter;

        @Override
        public void open(Configuration parameters) throws Exception {
            jedis = new Jedis("192.168.88.106", 6379);
            myBloomFilter = new MyBloomFilter(1 << 29); //要处理1亿个数据，用64MB大小的位图

        }

        @Override
        public void process(Context context, Iterable<UserBehavior> iterable, Collector<PageViewCount> collector) throws Exception {

            //将位图和窗口count值全部存入redis
            Long windowEnd = context.window().getEnd();
            String bitmapKey = windowEnd.toString();

            //把count值存成一张hash表
            String countHashName = "uv_count";
            String countKey = windowEnd.toString();

            //取当前的userId
            Long userId = iterable.iterator().next().getUserId();


            //计算位图中的offset

            Long offset = myBloomFilter.hashCode(userId.toString(), 61);

            //用redis的getbit命令，判断对应位置的值
            Boolean isExist = jedis.getbit(bitmapKey, offset);

            if (!isExist) {
                //如果不存在，在对应位图的位置，置1
                jedis.setbit(bitmapKey, offset, true);

                //更新redis中保存的count值
                Long uvCount = 0L;  //初始count值
                String uvCountString = jedis.hget(countHashName, countKey);

                if (uvCountString != null && !"".equals(uvCountString))
                    uvCount = Long.valueOf(uvCountString);
                jedis.hset(countHashName, countKey, String.valueOf(uvCount + 1));

                collector.collect(new PageViewCount("uv", windowEnd, uvCount + 1));
            }
        }

        @Override
        public void close() throws Exception {
            jedis.close();
        }
    }
}
